package org.apache.flink.web.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.client.cli.CliFrontend;
import org.apache.flink.client.cli.CustomCommandLine;
import org.apache.flink.client.cli.DefaultCLI;
import org.apache.flink.client.cli.GenericCLI;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.table.client.config.Environment;
import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.config.SqlClientOptions;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.context.DefaultContext;
import org.apache.flink.table.client.gateway.local.LocalExecutor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.File;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class CliConfig {

    private String flinkConfigDir;

    @Bean
    public Executor executor() {
        if (StringUtils.isBlank(this.flinkConfigDir)) {
            this.flinkConfigDir = CliFrontend.getConfigurationDirectoryFromEnv();
        }
        org.apache.flink.configuration.Configuration configuration = GlobalConfiguration.loadConfiguration(this.flinkConfigDir);
        configuration.set(SqlClientOptions.EXECUTION_RESULT_MODE, ResultMode.TABLEAU);

        List<CustomCommandLine> commandLines = new ArrayList<>();
        commandLines.add(new GenericCLI(configuration, this.flinkConfigDir));
        commandLines.add(new DefaultCLI());

        DefaultContext defaultContext = new DefaultContext(new Environment(), new ArrayList<>(), configuration, commandLines);
        final Executor executor = new LocalExecutor(defaultContext);
        executor.start();
        return executor;
    }

    @Value("${flink.configDir:}")
    public void setFlinkConfigDir(String flinkConfigDir) {
        if(StringUtils.isNoneBlank(flinkConfigDir)) {
            File file = new File(flinkConfigDir);
            if(file.exists()) {
                this.flinkConfigDir = flinkConfigDir;
            } else {
                if (flinkConfigDir.startsWith("classpath:")) {
                    flinkConfigDir = flinkConfigDir.substring("classpath:".length());
                }
                URL resource = getResource(flinkConfigDir);
                if (resource == null) {
                    throw new RuntimeException("flink目录配置不存在, 请配置flink.configDir");
                }
                System.out.println("resource:" + resource);
                this.flinkConfigDir = resource.getPath();
            }
        }
    }

    private URL getResource(String sourcePath) {
        for (ClassLoader loader : new ClassLoader[]{ClassLoader.getSystemClassLoader(), this.getClass().getClassLoader()}) {
            URL resource = loader.getResource(sourcePath);
            if(resource != null) {
                return resource;
            }
        }
        return null;
    }
}
